package com.lvmama.rhino.analyze.client

import com.lvmama.rhino.analyze.processing.{WirelessV2PitStat, WirelessV2Visitor, WirelessVer2Search}
import com.lvmama.rhino.common.entity.{JobConfig, KafkaTopic, TerminalType, WirelessClientMonitorLog}
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.kafka.{KafkaDStreamSource, KafkaPayloadStringCodec}
import com.lvmama.rhino.common.utils.sharedvariable.SessionAccumulator
import com.lvmama.rhino.common.utils.spark.StreamingApplication
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming._

/**
  * Created by yuanxiaofeng on 2017/4/18.
  */
class WirelessStatVer2(config: JobConfig, duration: Long) extends StreamingApplication{
  override def streamingBatchDuration: Long =  duration
  override var appName: String = "WirelessStatV2Job"
  override var sparkConfig: Map[String, String] = config.spark
  val streamingConfig: Map[String, String] = config.streamingCheckpointDir
  override def streamingCheckpointDir: String = streamingConfig
    .getOrElse("wirelessStatV2CP", "/checkpoint/wireless_v2")

  private val kafkaSourceTopic = KafkaTopic("wireless_v2")

  var kafkaConfig = config.sourceKafka

  def execute() = {
    sparkConfig += ("spark.sql.autoBroadcastJoinThreshold" -> "10485760")
//    sparkConfig += ("spark.master" -> "local[2]")
    sparkConfig += ("spark.app.name" -> appName)

    def func = (sc: SparkContext, ssc: StreamingContext) => {

      val streaming = KafkaDStreamSource(config.sourceKafka).createSource(ssc, kafkaSourceTopic.name)
//      val stringCodec = sc.broadcast(KafkaPayloadStringCodec())
      val lines = streaming
        .flatMap(KafkaPayloadStringCodec().decodeValue(_))
        .map(l => Utils.initWireless(l))
        .filter(l => l.logType != "")

      def trackStateFunc(batchTime: Time, key: String,
                         value: Option[WirelessClientMonitorLog],
                         state: State[String]
                        ): Option[(String, (WirelessClientMonitorLog, String))] = {
//        var sessionId = null.asInstanceOf[String]
//        val v = value.getOrElse((WirelessClientMonitorLog(), ""))
//        if (key == "") {
//          sessionId = v._1.sessionId
//        } else {
//          if (state.getOption().isDefined) {
//            sessionId = state.getOption().get
//          } else {
//            sessionId = java.util.UUID.randomUUID.toString.replace("-", "")
//          }
//        }
//        val output = (key, (v._1, v._2, sessionId))
//        state.update(sessionId)
//        Some(output)
        value match {
          case Some(v) =>
            val sessionId = state.getOption().getOrElse(java.util.UUID.randomUUID.toString.replace("-", ""))
            state.update(sessionId)
            Some((key, (value.get, sessionId)))
          case _ if state.isTimingOut() => Some(key, (WirelessClientMonitorLog(), ""))
        }
      }

      val statSpec = StateSpec.function(trackStateFunc _)
        .numPartitions(5)
        .timeout(Minutes(30))

      val wapDstream = lines
        .filter(r => r.platformCode == TerminalType.wab.toString)
        .map(r => (r.deviceToken, (r, r.sessionId)))
        .repartition(5)

      val wireless = lines
        .filter(r => r.platformCode != TerminalType.wab.toString)
        .map(r => (r.deviceToken, r))
        .mapWithState(statSpec)
        .union(wapDstream)
        .transform{ rdd =>
          rdd.map{r =>
            WirelessClientMonitorLog(r._2._1.logType, r._2._1.deviceToken, r._2._1.operator, r._2._1.netType,
              r._2._2, r._2._1.longitude, r._2._1.latitude, r._2._1.province, r._2._1.userSource, r._2._1.userId,
              r._2._1.buttonCode, r._2._1.pageCode, r._2._1.channelCode, r._2._1.platformCode, r._2._1.pageTypeCode, r._2._1.pageParam,
              r._2._1.timestamp, r._2._1.stationId, r._2._1.holeCode, r._2._1.accessCode, r._2._1.newVisitor, r._2._1.ip)
          }
        }

      new WirelessVer2Search().process(wireless)

      new WirelessV2PitStat().process(wireless)

      new WirelessV2Visitor().process(wireless)
    }

    val ssc = StreamingContext.getOrCreate(streamingCheckpointDir, () => withSparkStreamingContext(func))
    ssc.start()
    ssc.awaitTermination()
  }


}

object WirelessStatVer2 {
  def apply(duration: Long) = {
    val config = JobConfig()
    new WirelessStatVer2(config, duration).execute()
  }

  @volatile private var _sessionAcc: SessionAccumulator = null

  def getInstanceSessionAcc(sc: SparkContext): SessionAccumulator = {
    if (_sessionAcc == null) {
      synchronized {
        if (_sessionAcc == null) {
          _sessionAcc = new SessionAccumulator
          sc.register(_sessionAcc, "sessionAccumulator")
          _sessionAcc
        }
      }
    }
    _sessionAcc
  }

  @volatile private var _deviceBro: Broadcast[Map[Any, Any]] = null

  def getInstanceDeviceBro(sc:SparkContext, deviceCode: Map[Any, Any]): Broadcast[Map[Any, Any]] = {
    if (_deviceBro == null) {
      synchronized {
        if (_deviceBro == null) {
          _deviceBro = sc.broadcast(deviceCode)
        }
      }
    }
    _deviceBro
  }
}
